package com.xrui.hbase;

import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSortedSet;

import com.xrui.hbase.exception.HBaseURIException;
import com.xrui.hbase.util.NameValidator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;

import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URLDecoder;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * A {@link HBaseURI} that uniquely identifies an HBase instance, table, and column-set.
 *
 * <h2>{@code HBaseURI} Cluster Identifier</h2>
 * HBase needs only a valid ZooKeeper ensemble in order to identify the host HBase cluster.
 * The cluster identifier is identical to the default HBaseURI cluster identifier.
 *
 * The {@value #ENV_URI_STRING} value will resolve at runtime to the ZooKeeper ensemble address
 * in the HBase configuration on the classpath.
 *
 * <H2>{@code HBaseURI} Examples</H2>
 * The following are valid example {@code HBaseURI}s:
 *
 * <ul>
 * <li> {@code hbase://zkHost}
 * <li> {@code hbase://zkHost/instance}
 * <li> {@code hbase://zkHost/instance/table}
 * <li> {@code hbase://zkHost:zkPort/instance/table}
 * <li> {@code hbase://zkHost1,zkHost2/instance/table}
 * <li> {@code hbase://(zkHost1,zkHost2):zkPort/instance/table}
 * <li> {@code hbase://zkHost/instance/table/cf}
 * <li> {@code hbase://zkHost/instance/table/cf1,cf2}
 * <li> {@code hbase://.env/instance/table}
 * <li> {@code hbase://.unset/instance/table}
 * </ul>
 */
public final class HBaseURI {
    //region static
    /**
     * URI scheme used to fully qualify an HBase instance.
     */
    public static final String HBASE_SCHEME = "com/xrui/hbase";

    /**
     * String to specify an unset HBaseURI field.
     */
    public static final String UNSET_URI_STRING = ".unset";

    /**
     * String to specify a value through the local environment.
     */
    public static final String ENV_URI_STRING = ".env";

    /**
     * Default Zookeeper port.
     */
    public static final int DEFAULT_ZOOKEEPER_CLIENT_PORT = 2181;

    /**
     * Pattern matching "(host1,host2,host3):port".
     */
    public static final Pattern RE_AUTHORITY_GROUP = Pattern.compile("\\(([^)]+)\\):(\\d+)");
    /**
     * The scheme of this HBaseURI.
     */
    private final String mScheme;
    /**
     * Ordered list of Zookeeper quorum host names or IP addresses.
     * Preserves user ordering. Never null.
     */
    private final ImmutableList<String> mZookeeperQuorum;
    /**
     * Normalized (sorted) version of mZookeeperQuorum. Never null.
     */
    private final ImmutableList<String> mZookeeperQuorumNormalized;
    /**
     * Zookeeper client port number.
     */
    private final int mZookeeperClientPort;
    /**
     * HBase instance name. Null means unset.
     */
    private final String mInstanceName;
    /**
     * HBase table name. Null means unset.
     */
    private final String mTableName;

    //endregion

    //region variables
    /**
     * HBase column names. Never null. Empty means unset. Preserves user ordering.
     */
    private final ImmutableList<ColumnName> mColumnNames;
    /**
     * Normalized version of mColumnNames. Never null.
     */
    private final ImmutableList<ColumnName> mColumnNamesNormalized;
    /**
     * HBase query parameters. Null means unset.
     */
    ImmutableMap<String, String> mQueryParameters;
    /**
     * HBase action. Null means unset.
     */
    String mAction;

    /**
     * Constructs a new HBaseURI with the given parameters.
     *
     * @param scheme              of the URI.
     * @param zookeeperQuorum     Zookeeper quorum.
     * @param zookeeperClientPort Zookeeper client port.
     * @param instanceName        Instance name.
     * @param tableName           DatabaseTable name.
     * @param columnNames         Column names.
     * @throws HBaseURIException If the parameters are invalid.
     */
    private HBaseURI(
        final String scheme,
        final Iterable<String> zookeeperQuorum,
        final int zookeeperClientPort,
        final String instanceName,
        final String tableName,
        final Iterable<ColumnName> columnNames,
        final Map<String, String> queryParameters,
        final String action) {
        mScheme = scheme;
        mZookeeperQuorum = ImmutableList.copyOf(zookeeperQuorum);
        mZookeeperQuorumNormalized = ImmutableSortedSet.copyOf(mZookeeperQuorum).asList();
        mZookeeperClientPort = zookeeperClientPort;
        mInstanceName =
            ((null == instanceName) || !instanceName.equals(UNSET_URI_STRING)) ? instanceName : null;
        mTableName = ((null == tableName) || !tableName.equals(UNSET_URI_STRING)) ? tableName : null;
        mColumnNames = ImmutableList.copyOf(columnNames);
        mColumnNamesNormalized = ImmutableSortedSet.copyOf(columnNames).asList();
        mQueryParameters = ImmutableMap.copyOf(queryParameters);
        mAction = action;
        validateNames();
    }

    /**
     * Gets a builder configured with default HBase URI fields.
     * <p>
     * More precisely, the following defaults are initialized:
     * <ul>
     * <li>The Zookeeper quorum and client port is taken from the Hadoop <tt>Configuration</tt></li>
     * <li>The HBase instance name is set to <tt>KConstants.DEFAULT_INSTANCE_NAME</tt>
     * (<tt>"default"</tt>).</li>
     * <li>The table name and column names are explicitly left unset.</li>
     * </ul>
     *
     * @return A builder configured with this HBase URI.
     */
    public static HBaseURIBuilder newBuilder() {
        return new HBaseURIBuilder();
    }

    /**
     * Gets a builder configured with a HBase URI.
     *
     * @param uri The HBase URI to configure the builder from.
     * @return A builder configured with uri.
     */
    public static HBaseURIBuilder newBuilder(HBaseURI uri) {
        return new HBaseURIBuilder(
            uri.getScheme(),
            uri.getZookeeperQuorumOrdered(),
            uri.getZookeeperClientPort(),
            uri.getInstance(),
            uri.getTable(),
            uri.getColumnsOrdered(),
            uri.getParameters(),
            uri.getAction());
    }

    /**
     * Gets a builder configured with the HBase URI.
     *
     * The String parameter can be a relative URI (with a specified instance), in which
     * case it is automatically normalized relative to DEFAULT_HBASE_URI.
     *
     * @param uri String specification of a HBase URI.
     * @return A builder configured with uri.
     * @throws HBaseURIException If the uri is invalid.
     */
    public static HBaseURIBuilder newBuilder(final String uri) {
        final String uriWithScheme;
        if (uri.startsWith(HBASE_SCHEME)) {
            uriWithScheme = uri;
        } else {
            uriWithScheme = String.format("%s/%s/", Constants.DEFAULT_HBASE_URI, uri);
        }
        try {
            return new HBaseURIParser().parse(new URI(uriWithScheme));
        } catch (URISyntaxException exn) {
            throw new HBaseURIException(uri, exn.getMessage());
        }
    }

    /**
     * Gets a HBase URI from a string specification.
     *
     * The String parameter can be a relative URI (with a specified instance), in which
     * case it is automatically normalized relative to DEFAULT_HBASE_URI.
     *
     * @param uri String specification of a HBase URI.
     * @return A HBase uri.
     * @throws HBaseURIException If the uri is invalid.
     */
    public static HBaseURI ofValue(final String uri) {
        return newBuilder(uri).build();
    }

    /**
     * Returns the set of Zookeeper quorum hosts (names or IPs).
     *
     * Host names or IP addresses are de-duplicated and sorted.
     *
     * @return the set of Zookeeper quorum hosts (names or IPs).
     * Never null.
     */
    public ImmutableList<String> getZookeeperQuorum() {
        return mZookeeperQuorumNormalized;
    }
    //endregion

    //region private construct, only can create instance using the HBaseURIBuilder or HBaseURI.ofValue()

    /**
     * Returns the original user-specified list of Zookeeper quorum hosts.
     *
     * <p> Host names are exactly as specified by the user. </p>
     *
     * @return the original user-specified list of Zookeeper quorum hosts.
     * Never null.
     */
    public ImmutableList<String> getZookeeperQuorumOrdered() {
        return mZookeeperQuorum;
    }
    //endregion

    //region APIs

    /**
     * @return Zookeeper client port.
     */
    public int getZookeeperClientPort() {
        return mZookeeperClientPort;
    }

    /**
     * Returns the scheme of this HBaseURI.
     *
     * @return the scheme of this HBaseURI.
     */
    public String getScheme() {
        return mScheme;
    }

    /**
     * Returns the name of the HBase instance specified by this URI, if any.
     *
     * @return the name of the HBase instance specified by this URI.
     * Null means unspecified (ie. this URI does not target a HBase instance).
     */
    public String getInstance() {
        return mInstanceName;
    }

    /**
     * Returns the name of the HBase table specified by this URI, if any.
     *
     * @return the name of the HBase table specified by this URI.
     * Null means unspecified (ie. this URI does not target a HBase table).
     */
    public String getTable() {
        return mTableName;
    }

    /**
     * @return HBase column names, normalized. Never null.
     */
    public ImmutableList<ColumnName> getColumnNames() {
        return mColumnNamesNormalized;
    }

    /**
     * @return HBase columns, ordered. Never null.
     */
    public Collection<ColumnName> getColumnsOrdered() {
        return mColumnNames;
    }

    /**
     * @return Parameters. Null means unspecified.
     */
    public Map<String, String> getParameters() {
        return mQueryParameters;
    }

    /**
     * Returns the action name of the HBase table specified by this URI, if any.
     *
     * @return the action name of the HBase table specified by this URI.
     * Null means unspecified (ie. this URI does not target a HBase table action).
     */
    public String getAction() {
        return mAction;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return toString(false);
    }

    /**
     * Returns a string representation of this URI that preserves ordering of lists in fields,
     * such as the Zookeeper quorum and HBase columns.
     *
     * @return An order-preserving string representation of this URI.
     */
    public String toOrderedString() {
        return toString(true);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return toString().hashCode();
    }

    /**
     * {@inheritDoc}
     * <p>
     * *note* {@code HBaseURI}'s are compared based on the string representation of the {@code URI}.
     * If two {@code HBaseURI}s point to the same cluster with a different scheme, then they will
     * compare differently, for example {@code hbase://localhost} and {@code hbase://localhost}.
     */
    @Override
    public boolean equals(Object object) {
        return object != null && object.getClass() == this.getClass() && object.toString().equals(this.toString());
    }

    /**
     * Overridden to provide specific return type.
     * <p>
     * {@inheritDoc}
     */
    public HBaseURI resolve(String path) {
        try {
            // Without the "./", URI will assume a path containing a colon
            // is a new URI, for example "family:column".
            URI uri = new URI(toString()).resolve(String.format("./%s", path));
            return newBuilder(uri.toString()).build();
        } catch (URISyntaxException e) {
            // This should never happen
            throw new InternalError(String.format("HBaseURI was incorrectly constructed: %s.", this));
        } catch (IllegalArgumentException e) {
            throw new HBaseURIException(this.toString(),
                String.format("Path can not be resolved: %s", path));
        }
    }

    /**
     * Returns a string representation of this URI.
     *
     * @param preserveOrdering Whether to preserve ordering of lsits in fields.
     * @return A string representation of this URI.
     */
    private String toString(boolean preserveOrdering) {
        // Remove trailing unset fields.
        if (!mColumnNames.isEmpty()) {
            return toStringCol(preserveOrdering);
        } else if (mTableName != null) {
            return toStringTable(preserveOrdering);
        } else if (mInstanceName != null) {
            return toStringInstance(preserveOrdering);
        } else {
            return toStringAuthority(preserveOrdering);
        }
    }

    /**
     * Validates the names used in the URI.
     *
     * @throws HBaseURIException if there is an invalid name in this URI.
     */
    private void validateNames() {
        if ((mInstanceName != null) && !NameValidator.isValidName(mInstanceName)) {
            throw new HBaseURIException(String.format(
                "Invalid HBase URI: '%s' is not a valid HBase instance name.", mInstanceName));
        }
        if ((mTableName != null) && !NameValidator.isValidTableName(mTableName)) {
            throw new HBaseURIException(String.format(
                "Invalid HBase URI: '%s' is not a valid HBase table name.", mTableName));
        }
    }

    //endregion

    //region private or protect methods

    /**
     * Appends the cluster identifier for this {@code HBaseURI} to the passed in {@link StringBuilder}.
     * <p>
     * This is a default implementation which will append the ZooKeeper ensemble to the string
     * builder.  May be overridden by subclasses if more than the ZooKeeper ensemble is necessary for
     * the cluster identifier.
     *
     * @param sb               a StringBuilder to append the cluster identifier to.
     * @param preserveOrdering whether to preserve ordering in the cluster identifier components.
     */
    private void appendClusterIdentifier(
        final StringBuilder sb,
        final boolean preserveOrdering
    ) {
        ImmutableList<String> zookeeperQuorum =
            preserveOrdering ? mZookeeperQuorum : mZookeeperQuorumNormalized;
        if (zookeeperQuorum == null) {
            sb.append(UNSET_URI_STRING);
        } else if (zookeeperQuorum.size() == 1) {
            sb.append(zookeeperQuorum.get(0));
        } else {
            sb.append('(');
            Joiner.on(',').appendTo(sb, zookeeperQuorum);
            sb.append(')');
        }
        sb
            .append(':')
            .append(mZookeeperClientPort)
            .append('/');

    }

    /**
     * Formats the full HBaseURI up to the authority, preserving order.
     *
     * @param preserveOrdering Whether to preserve ordering.
     * @return Representation of this HBaseURI up to the authority.
     */
    private String toStringAuthority(boolean preserveOrdering) {
        StringBuilder sb = new StringBuilder();
        sb.append(mScheme)
            .append("://");

        appendClusterIdentifier(sb, preserveOrdering);
        return sb.toString();
    }

    /**
     * Formats the full HBaseURI up to the instance.
     *
     * @param preserveOrdering Whether to preserve ordering.
     * @return Representation of this HBaseURI up to the instance.
     */
    private String toStringInstance(boolean preserveOrdering) {
        return String.format("%s%s/",
            toStringAuthority(preserveOrdering),
            (null == mInstanceName) ? UNSET_URI_STRING : mInstanceName);
    }

    /**
     * Formats the full HBaseURI up to the table.
     *
     * @param preserveOrdering Whether to preserve ordering.
     * @return Representation of this HBaseURI up to the table.
     */
    private String toStringTable(boolean preserveOrdering) {
        return String.format("%s%s/",
            toStringInstance(preserveOrdering),
            (null == mTableName) ? UNSET_URI_STRING : mTableName);
    }

    /**
     * Formats the full HBaseURI up to the column.
     *
     * @param preserveOrdering Whether to preserve ordering.
     * @return Representation of this HBaseURI up to the table.
     */
    private String toStringCol(boolean preserveOrdering) {
        String columnField;
        ImmutableList<ColumnName> columnNames =
            preserveOrdering ? mColumnNames : mColumnNamesNormalized;
        if (columnNames.isEmpty()) {
            columnField = UNSET_URI_STRING;
        } else {
            ImmutableList.Builder<String> builder = ImmutableList.builder();
            for (ColumnName cn : columnNames) {
                builder.add(cn.toString());
            }
            ImmutableList<String> strColumns = builder.build();
            if (strColumns.size() == 1) {
                columnField = strColumns.get(0);
            } else {
                columnField = Joiner.on(",").join(strColumns);
            }
        }

        try {
            // SCHEMA-6. URI Encode column names using RFC-2396.
            final URI columnsEncoded = new URI(HBASE_SCHEME, columnField, null);
            return String.format("%s%s/",
                toStringTable(preserveOrdering),
                columnsEncoded.getRawSchemeSpecificPart());
        } catch (URISyntaxException e) {
            throw new HBaseURIException(e.getMessage());
        }
    }

    protected HBaseURIBuilder getBuilder() {
        return newBuilder(this);
    }

    /**
     * Builder class for constructing HBaseURIs.
     */
    public static final class HBaseURIBuilder {
        /**
         * The scheme of the HBaseURI being built.
         */
        String mScheme;

        /**
         * Zookeeper quorum: comma-separated list of Zookeeper host names or IP addresses.
         * Preserves user ordering.
         */
        ImmutableList<String> mZookeeperQuorum;

        /**
         * Zookeeper client port number.
         */
        int mZookeeperClientPort;

        /**
         * HBase instance name. Null means unset.
         */
        String mInstanceName;

        /**
         * HBase table name. Null means unset.
         */
        String mTableName;

        /**
         * HBase column names. Never null. Empty means unset. Preserves user ordering.
         */
        ImmutableList<ColumnName> mColumnNames;

        /**
         * HBase query parameters. Null means unset.
         */
        ImmutableMap<String, String> mQueryParameters;

        /**
         * HBase action. Null means unset.
         */
        String mAction;

        /**
         * Constructs a new builder for HBaseURIs.
         *
         * @param scheme              of the URI.
         * @param zookeeperQuorum     The initial zookeeper quorum.
         * @param zookeeperClientPort The initial zookeeper client port.
         * @param instanceName        The initial instance name.
         * @param tableName           The initial table name.
         * @param columnNames         The initial column names.
         * @param queryParameters     The initial query parameters.
         * @param action              The initial action.
         */
        private HBaseURIBuilder(
            final String scheme,
            final Iterable<String> zookeeperQuorum,
            final int zookeeperClientPort,
            final String instanceName,
            final String tableName,
            final Iterable<ColumnName> columnNames,
            final Map<String, String> queryParameters,
            final String action
        ) {
            mScheme = scheme;
            mZookeeperQuorum = ImmutableList.copyOf(zookeeperQuorum);
            mZookeeperClientPort = zookeeperClientPort;
            mInstanceName =
                ((null == instanceName) || !instanceName.equals(UNSET_URI_STRING)) ? instanceName : null;
            mTableName = ((null == tableName) || !tableName.equals(UNSET_URI_STRING)) ? tableName : null;
            mColumnNames = ImmutableList.copyOf(columnNames);
            mQueryParameters = ImmutableMap.copyOf(queryParameters);
            mAction = action;
        }

        /**
         * Constructs a new builder for HBaseURIs.
         */
        public HBaseURIBuilder() {
            mScheme = HBASE_SCHEME;
            mZookeeperQuorum = HBaseURIParser.ZooKeeperAuthorityParser.ENV_ZOOKEEPER_QUORUM;
            mZookeeperClientPort = HBaseURIParser.ZooKeeperAuthorityParser.ENV_ZOOKEEPER_CLIENT_PORT;
            mInstanceName = Constants.DEFAULT_INSTANCE_NAME;
            mTableName = UNSET_URI_STRING;
            mColumnNames = ImmutableList.of();
            mQueryParameters = ImmutableMap.of();
        }

        /**
         * Configures the HBaseURI with Zookeeper Quorum.
         *
         * @param zookeeperQuorum The zookeeper quorum.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder withZookeeperQuorum(String[] zookeeperQuorum) {
            mZookeeperQuorum = ImmutableList.copyOf(zookeeperQuorum);
            return this;
        }

        /**
         * Configures the HBaseURI with Zookeeper Quorum.
         *
         * @param zookeeperQuorum The zookeeper quorum.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder withZookeeperQuorum(Iterable<String> zookeeperQuorum) {
            mZookeeperQuorum = ImmutableList.copyOf(zookeeperQuorum);
            return this;
        }

        /**
         * Configures the HBaseURI with the Zookeeper client port.
         *
         * @param zookeeperClientPort The port.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder withZookeeperClientPort(int zookeeperClientPort) {
            mZookeeperClientPort = zookeeperClientPort;
            return this;
        }

        /**
         * Configures the HBaseURI with the HBase instance name.
         *
         * @param instanceName The HBase instance name.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder withInstanceName(String instanceName) {
            mInstanceName = instanceName;
            return this;
        }

        /**
         * Configures the HBaseURI with the HBase table name.
         *
         * @param tableName The HBase table name.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder withTableName(String tableName) {
            mTableName = tableName;
            return this;
        }

        /**
         * Configures the HBaseURI with the HBase column names.
         *
         * @param columnNames The HBase column names to configure.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder withColumnNames(Collection<ColumnName> columnNames) {
            ImmutableList.Builder<ColumnName> builder = ImmutableList.builder();
            for (ColumnName cn : columnNames) {
                builder.add(cn);
            }
            mColumnNames = builder.build();
            return this;
        }

        /**
         * Adds the column names to the HBase URI column names.
         *
         * @param columnNames The HBase column names to add.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder addColumnNames(Collection<ColumnName> columnNames) {
            ImmutableList.Builder<ColumnName> builder = ImmutableList.builder();
            builder.addAll(mColumnNames).addAll(columnNames);
            mColumnNames = builder.build();
            return this;
        }

        /**
         * Adds the column name to the HBase URI column names.
         *
         * @param columnName The HBase column name to add.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder addColumnName(ColumnName columnName) {
            ImmutableList.Builder<ColumnName> builder = ImmutableList.builder();
            builder.addAll(mColumnNames).add(columnName);
            mColumnNames = builder.build();
            return this;
        }

        /**
         * Configures the HBaseURI with the query parameters.
         *
         * @param queryParameters The query parameters to configure.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder withQueryParameters(Map<String, String> queryParameters) {
            ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
            builder.putAll(queryParameters);
            mQueryParameters = builder.build();
            return this;
        }

        /**
         * Adds the query parameters.
         *
         * @param queryParameters The query parameters to add.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder addQueryParameters(Map<String, String> queryParameters) {
            ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
            builder.putAll(mQueryParameters).putAll(queryParameters);
            mQueryParameters = builder.build();
            return this;
        }

        /**
         * Adds the query parameter to the query parameters.
         *
         * @param key   The parameter name to add.
         * @param value The parameter value to add.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder addQueryParameter(String key, String value) {
            ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
            builder.putAll(mQueryParameters).put(key, value);
            mQueryParameters = builder.build();
            return this;
        }

        /**
         * Configures the HBaseURI with the action name.
         *
         * @param action The action name.
         * @return This builder instance so you may chain configuration method calls.
         */
        public HBaseURIBuilder withAction(String action) {
            mAction = action;
            return this;
        }

        /**
         * Builds the configured HBaseURI.
         *
         * @return A HBaseURI.
         * @throws HBaseURIException If the HBaseURI was configured improperly.
         */
        public HBaseURI build() {
            return new HBaseURI(
                mScheme,
                mZookeeperQuorum,
                mZookeeperClientPort,
                mInstanceName,
                mTableName,
                mColumnNames,
                mQueryParameters,
                mAction);
        }
    }

    /**
     * A {@link HBaseURIParser} for {@link HBaseURI}s.
     */
    public static final class HBaseURIParser {
        /**
         * Parse a HBaseURI and return a {@link HBaseURIBuilder}.
         *
         * @param uri to parse.
         * @return a URIBuilder parsed from the HBaseURI.
         */
        public HBaseURIBuilder parse(final URI uri) {
            final ZooKeeperAuthorityParser authorityParser = ZooKeeperAuthorityParser.getAuthorityParser(uri);
            final PathParser segmentParser = new PathParser(uri);
            final QueryParameterParser queryParser = new QueryParameterParser(uri);

            return new HBaseURIBuilder(
                uri.getScheme(),
                authorityParser.getZookeeperQuorum(),
                authorityParser.getZookeeperClientPort(),
                segmentParser.getInstance(),
                segmentParser.getTable(),
                segmentParser.getColumnNames(),
                queryParser.getQueryParameters(),
                uri.getFragment());
        }

        public String getName() {
            return HBASE_SCHEME;
        }

        /**
         * Class for parsing an authority portion in a HBaseURI.
         */
        static final class AuthorityParser {
            private final ImmutableList<String> mHosts;
            private final Integer mHostPort;

            /**
             * Return a new authority parser, which will parse a substring of the entire HBaseURI.
             * <p>
             * We specify the authority explicitly because in a Cassandra HBaseURI, the Cassandra hosts
             * authority will be different from `uri.getAuthority()`.
             * <p>
             * Parse the hosts from a host / port segment. The segment must be in one of the following
             * formats:
             * <p>
             * <li> {@code host}
             * <li> {@code host:port}
             * <li> {@code host1,host2}
             * <li> {@code (host1,host2):port}
             *
             * @param authority The substring of the HBaseURI containing the authority.
             * @param uri       The HBaseURI itself (needed here for error messages).
             */
            private AuthorityParser(String authority, URI uri) {
                // Extract the user info, if it is present.
                final String hostPortInfo = getHostAndPortInfoFromAuthority(uri, authority);

                // Get the hosts and ports.
                mHosts = parseHosts(uri, hostPortInfo);
                mHostPort = parsePort(uri, hostPortInfo);
            }

            /**
             * Constructs an AuthorityParser.
             *
             * @param authority The authority within the HBaseURI.
             * @param uri       The uri whose authority is to be parsed (used for generating messages in
             *                  exceptions).
             * @return an {@code AuthorityParser} for the provided {@code HBaseURI}.
             * @throws HBaseURIException If the authority is invalid.
             */
            public static AuthorityParser getAuthorityParser(String authority, URI uri) {
                return new AuthorityParser(authority, uri);
            }

            /**
             * Extract the host(s) and port from an authority.
             * <p>
             * The returned value is of the form `host`, `host:port`, `(host1,host2,...)`, or
             * `(host1,host2,...):port`.
             *
             * @param uri       The HBaseURI containing the authority (used to generate exception messages).
             * @param authority The authority to parse.
             * @return A string containing the host(s) and optional port.
             */
            private static String getHostAndPortInfoFromAuthority(URI uri, String authority) {
                if (!authority.contains("@")) {
                    return authority;
                }
                // Should have no more than a single @ appear!
                final List<String> userInfoAndHostPortInfo =
                    ImmutableList.copyOf(Splitter.on('@').split(authority));
                if (2 != userInfoAndHostPortInfo.size()) {
                    throw new HBaseURIException(
                        uri.toString(), "Cannot have more than one '@' in HBaseURI authority");
                }
                Preconditions.checkArgument(2 == userInfoAndHostPortInfo.size());
                return userInfoAndHostPortInfo.get(1);
            }

            /**
             * Parse the hosts from a host / port segment. The segment must be in one of the following
             * formats:
             * <p>
             * <li> {@code host}
             * <li> {@code host:port}
             * <li> {@code host1,host2}
             * <li> {@code (host1,host2):port}
             *
             * @param uri     being parsed. Used for error messaging.
             * @param segment containing the host(s) and optionally the port.
             * @return list of hosts.
             */
            private static ImmutableList<String> parseHosts(URI uri, String segment) {
                final Matcher zkMatcher = HBaseURI.RE_AUTHORITY_GROUP.matcher(segment);
                if (zkMatcher.matches()) {
                    return ImmutableList.copyOf(Splitter.on(',').split(zkMatcher.group(1)));
                } else {
                    final List<String> splits =
                        ImmutableList.copyOf(Splitter.on(':').omitEmptyStrings().split(segment));
                    switch (splits.size()) {
                        case 1:
                            return ImmutableList.copyOf(Splitter.on(',').split(segment));
                        case 2:
                            if (splits.get(0).contains(",")) {
                                throw new HBaseURIException(
                                    uri.toString(), "Multiple ZooKeeper hosts must be parenthesized.");
                            }
                            return ImmutableList.of(splits.get(0));
                        default:
                            throw new HBaseURIException(
                                uri.toString(), "Invalid ZooKeeper ensemble cluster identifier.");
                    }
                }
            }

            /**
             * Parse the port from a host / port segment. The segment must be in one of the following
             * formats:
             * <p>
             * <li> {@code host}
             * <li> {@code host:port}
             * <li> {@code host1,host2}
             * <li> {@code (host1,host2):port}
             *
             * @param uri     being parsed. Used for error messaging.
             * @param segment containing the host(s) and optionally the port.
             * @return the port, or {@code null} if no port is specified.
             */
            private static Integer parsePort(URI uri, String segment) {
                final List<String> parts = ImmutableList.copyOf(Splitter.on(':').split(segment));
                if (parts.size() != 2) {
                    return null;
                }
                try {
                    return Integer.parseInt(parts.get(1));
                } catch (NumberFormatException nfe) {
                    throw new HBaseURIException(uri.toString(),
                        String.format("Can not parse port '%s'.", parts.get(1)));
                }
            }

            /**
             * @return the hosts from the parsed HBaseURI authority.
             */
            public ImmutableList<String> getHosts() {
                return mHosts;
            }

            /**
             * @return the host port from the parsed HBaseURI authority (can be null).
             */
            public Integer getHostPort() {
                return mHostPort;
            }
        }

        /**
         * Parser for the ZooKeeper authority section in a HBaseURI.
         */
        static final class ZooKeeperAuthorityParser {
            /**
             * ZooKeeper quorum configured from the local environment.
             */
            public static final ImmutableList<String> ENV_ZOOKEEPER_QUORUM;

            /**
             * ZooKeeper client port configured from the local environment.
             */
            public static final int ENV_ZOOKEEPER_CLIENT_PORT;

            /**
             * Resolves the local environment ZooKeeper parameters.
             *
             * Local environment refers to the hbase-site.xml configuration file available on the classpath.
             */
            static {
                Configuration conf = HBaseConfiguration.create();
                ENV_ZOOKEEPER_QUORUM = ImmutableList.copyOf(conf.get(HConstants.ZOOKEEPER_QUORUM).split(","));
                ENV_ZOOKEEPER_CLIENT_PORT =
                    conf.getInt(HConstants.ZOOKEEPER_CLIENT_PORT, HBaseURI.DEFAULT_ZOOKEEPER_CLIENT_PORT);
            }

            /**
             * List of nodes in the ZooKeeper quorum for this authority.
             */
            private final ImmutableList<String> mZookeeperQuorum;

            /**
             * The ZooKeeper port for this authority.
             */
            private final int mZookeeperClientPort;

            /**
             * Constructs an AuthorityParser for ZooKeeper authorities.
             *
             * @param uri The uri whose authority is to be parsed.
             * @throws HBaseURIException If the authority is invalid.
             */
            private ZooKeeperAuthorityParser(URI uri) {
                final String authority = uri.getAuthority();
                if (null == authority) {
                    throw new HBaseURIException(uri.toString(), "ZooKeeper ensemble missing.");
                }

                if (authority.equals(HBaseURI.ENV_URI_STRING)) {
                    mZookeeperQuorum = ENV_ZOOKEEPER_QUORUM;
                    mZookeeperClientPort = ENV_ZOOKEEPER_CLIENT_PORT;
                } else {
                    final AuthorityParser authorityParser = AuthorityParser.getAuthorityParser(authority, uri);
                    mZookeeperQuorum = authorityParser.getHosts();
                    final Integer port = authorityParser.getHostPort();
                    mZookeeperClientPort = port == null ? HBaseURI.DEFAULT_ZOOKEEPER_CLIENT_PORT : port;
                }
            }

            /**
             * Constructs an AuthorityParser.
             *
             * @param uri The uri whose authority is to be parsed.
             * @return an {@code AuthorityParser} for the provided {@code HBaseURI}.
             * @throws HBaseURIException If the authority is invalid.
             */
            public static ZooKeeperAuthorityParser getAuthorityParser(URI uri) {
                return new ZooKeeperAuthorityParser(uri);
            }

            /**
             * Gets the zookeeper quorum.
             *
             * @return The zookeeper quorum.
             */
            public ImmutableList<String> getZookeeperQuorum() {
                return mZookeeperQuorum;
            }

            /**
             * Gets the zookeeper client port.
             *
             * @return The zookeeper client port.
             */
            public int getZookeeperClientPort() {
                return mZookeeperClientPort;
            }
        }

        /**
         * Parses {@code HBaseURI} path segments into components.
         */
        public static class PathParser {

            private final String mInstance;
            private final String mTable;
            private final List<ColumnName> mColumnNames;

            /**
             * Create a new path segment parser for a HBaseURI.
             *
             * @param uri to parse.
             */
            public PathParser(URI uri) {
                this(uri, 0);
            }

            /**
             * Create a new path segment parser for a HBaseURI with the provided instanceIndex.  The HBase
             * components begin at the {@code instanceIndex} offset of the path segments.
             *
             * @param uri           to parse.
             * @param instanceIndex of HBase component.
             */
            public PathParser(URI uri, int instanceIndex) {
                final List<String> segments =
                    ImmutableList.copyOf(Splitter.on('/').omitEmptyStrings().split(uri.getPath()));
                final int size = segments.size();

                if (size - instanceIndex > 3) {
                    throw new HBaseURIException(uri.toString(), "Too many path segments.");
                }
                final int tableIndex = instanceIndex + 1;
                final int columnsIndex = tableIndex + 1;

                if (size > instanceIndex && !segments.get(instanceIndex).equals(HBaseURI.UNSET_URI_STRING)) {
                    mInstance = segments.get(instanceIndex);
                } else {
                    mInstance = null;
                }

                if (size > tableIndex && !segments.get(tableIndex).equals(HBaseURI.UNSET_URI_STRING)) {
                    mTable = segments.get(tableIndex);
                } else {
                    mTable = null;
                }

                if (size > columnsIndex && !segments.get(columnsIndex).equals(HBaseURI.UNSET_URI_STRING)) {
                    ImmutableList.Builder<ColumnName> columns = ImmutableList.builder();
                    for (String column
                        : Splitter.on(',').omitEmptyStrings().split(segments.get(columnsIndex))) {
                        columns.add(ColumnName.create(column));
                    }
                    mColumnNames = columns.build();
                } else {
                    mColumnNames = ImmutableList.of();
                }
            }

            /**
             * Get the parsed instance name.
             *
             * @return the instance name.
             */
            public String getInstance() {
                return mInstance;
            }

            /**
             * Get the parsed table name.
             *
             * @return the table name.
             */
            public String getTable() {
                return mTable;
            }

            /**
             * Get the column names.
             *
             * @return the column names.
             */
            public List<ColumnName> getColumnNames() {
                return mColumnNames;
            }
        }

        /**
         * Parser for the query parameters section in a HBaseURI.
         */
        public static final class QueryParameterParser {
            /**
             * Map of query parameters.
             */
            private final ImmutableMap<String, String> mQueryParameters;

            /**
             * Constructs an QueryParameterParser for table actions.
             *
             * @param uri The uri whose query parameters are to be parsed.
             * @throws HBaseURIException If the query string is invalid.
             */
            private QueryParameterParser(URI uri) {
                final String queryString = uri.getQuery();
                ImmutableMap.Builder<String, String> parameters = ImmutableMap.builder();
                if (queryString != null && queryString.length() > 0) {
                    int ampersandIndex, lastAmpersandIndex = 0;
                    String subStr, name, value = null;
                    String[] paramPair;
                    do {
                        ampersandIndex = queryString.indexOf('&', lastAmpersandIndex) + 1;
                        if (ampersandIndex > 0) {
                            subStr = queryString.substring(lastAmpersandIndex, ampersandIndex - 1);
                            lastAmpersandIndex = ampersandIndex;
                        } else {
                            subStr = queryString.substring(lastAmpersandIndex);
                        }
                        paramPair = subStr.split("=");
                        name = paramPair[0];
                        if (paramPair.length != 1) {
                            try {
                                value = URLDecoder.decode(paramPair[1], "utf-8");
                            } catch (UnsupportedEncodingException ignored) {
                            }
                        }
                        parameters.put(name, value);
                    } while (ampersandIndex > 0);
                }
                mQueryParameters = parameters.build();
            }

            /**
             * Constructs an QueryParameterParser.
             *
             * @param uri The uri whose query parameters are to be parsed.
             * @return an {@code QueryParameterParser} for the provided {@code HBaseURI}.
             * @throws HBaseURIException If the authority is invalid.
             */
            public static QueryParameterParser getQueryParameterParser(URI uri) {
                return new QueryParameterParser(uri);
            }

            /**
             * Gets the query parameters.
             *
             * @return The query parameters.,
             */
            public ImmutableMap<String, String> getQueryParameters() {
                return mQueryParameters;
            }

            /**
             * Gets the specified query parameter value.
             *
             * @return The value of the specified parameter or null if parameter not specified.
             */
            public String getQueryParameter(String name) {
                return mQueryParameters.getOrDefault(name, null);
            }
        }
    }
    //endregion
}
